package org.jetlinks.protocol.official;

import com.fasterxml.jackson.databind.ObjectMapper;
import io.netty.buffer.Unpooled;
import lombok.extern.slf4j.Slf4j;
import org.jetlinks.core.message.DeviceMessage;
import org.jetlinks.core.message.DisconnectDeviceMessage;
import org.jetlinks.core.message.Message;
import org.jetlinks.core.message.codec.*;
import org.jetlinks.core.message.function.FunctionInvokeMessage;
import org.jetlinks.protocol.official.mqtt.TopicMatcher;
import org.jetlinks.protocol.official.mqtt.down.CacheKeyEnum;
import org.jetlinks.protocol.official.mqtt.down.DownTopicMessageEnum;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import javax.annotation.Nonnull;
import java.util.Arrays;

/**
 * <pre>
 *     下行Topic:
 *          读取设备属性: /{productId}/{deviceId}/properties/read
 *          修改设备属性: /{productId}/{deviceId}/properties/write
 *          调用设备功能: /{productId}/{deviceId}/function/invoke
 *
 *          //网关设备
 *          读取子设备属性: /{productId}/{deviceId}/child/{childDeviceId}/properties/read
 *          修改子设备属性: /{productId}/{deviceId}/child/{childDeviceId}/properties/write
 *          调用子设备功能: /{productId}/{deviceId}/child/{childDeviceId}/function/invoke
 *
 *      上行Topic:
 *          读取属性回复: /{productId}/{deviceId}/properties/read/reply
 *          修改属性回复: /{productId}/{deviceId}/properties/write/reply
 *          调用设备功能: /{productId}/{deviceId}/function/invoke/reply
 *          上报设备事件: /{productId}/{deviceId}/event/{eventId}
 *          上报设备属性: /{productId}/{deviceId}/properties/report
 *          上报设备派生物模型: /{productId}/{deviceId}/metadata/derived
 *
 *          //网关设备
 *          子设备上线消息: /{productId}/{deviceId}/child/{childDeviceId}/connected
 *          子设备下线消息: /{productId}/{deviceId}/child/{childDeviceId}/disconnect
 *          读取子设备属性回复: /{productId}/{deviceId}/child/{childDeviceId}/properties/read/reply
 *          修改子设备属性回复: /{productId}/{deviceId}/child/{childDeviceId}/properties/write/reply
 *          调用子设备功能回复: /{productId}/{deviceId}/child/{childDeviceId}/function/invoke/reply
 *          上报子设备事件: /{productId}/{deviceId}/child/{childDeviceId}/event/{eventId}
 *          上报子设备派生物模型: /{productId}/{deviceId}/child/{childDeviceId}/metadata/derived
 *
 * </pre>
 * 基于jet links 的消息编解码器
 *
 * @author zhouhao
 * @since 1.0.0
 */
@Slf4j
public class JetLinksMqttDeviceMessageCodec implements DeviceMessageCodec {

    private final Transport transport;

    private final ObjectMapper mapper;

    public JetLinksMqttDeviceMessageCodec(Transport transport) {
        this.transport = transport;
        this.mapper = ObjectMappers.JSON_MAPPER;
    }

    public JetLinksMqttDeviceMessageCodec() {
        this(DefaultTransport.MQTT);
    }

    @Override
    public Transport getSupportTransport() {
        return transport;
    }

    @Nonnull
    public Mono<MqttMessage> encode(@Nonnull MessageEncodeContext context) {
        return Mono.defer(() -> {
            Message message = context.getMessage();

            if (message instanceof DisconnectDeviceMessage) {
                return ((ToDeviceMessageContext) context)
                        .disconnect()
                        .then(Mono.empty());
            }

            if (message instanceof DeviceMessage) {
                FunctionInvokeMessage functionInvokeMessage = ((FunctionInvokeMessage) message);
                String functionId = functionInvokeMessage.getFunctionId();

                // 根据functionId前缀匹配需要调用的产品功能
                DownTopicMessageEnum downTopicMessageEnum = Arrays.stream(DownTopicMessageEnum.values())
                                                                  .filter(item -> functionId.startsWith(item.getFunctionIdPrefix()))
                                                                  .findFirst()
                                                                  .orElseThrow(() -> new UnsupportedOperationException("不支持的functionId:" + functionId));
                TopicPayload topicPayload = downTopicMessageEnum
                        .getFunction()
                        .apply(functionInvokeMessage, downTopicMessageEnum.getTopicPatten());

                if (topicPayload == null) {
                    return Mono.empty();
                }
                String key = CacheKeyEnum.FunctionId_.name() + functionInvokeMessage.getDeviceId();
                return context.getDevice(functionInvokeMessage.getDeviceId())
                        .flatMap(deviceOperator -> deviceOperator.setConfig(key, functionInvokeMessage.getMessageId())
                                .thenReturn(SimpleMqttMessage
                                                    .builder()
                                                    .clientId(functionInvokeMessage.getDeviceId())
                                                    .topic(topicPayload.getTopic())
                                                    .payloadType(MessagePayloadType.JSON)
                                                    .payload(Unpooled.wrappedBuffer(topicPayload.getPayload()))
                                                    .build()));
            } else {
                return Mono.empty();
            }
        });
    }

    @Nonnull
    @Override
    public Publisher<DeviceMessage> decode(@Nonnull MessageDecodeContext context) {

        MqttMessage message = (MqttMessage) context.getMessage();
        log.warn("bingzhong decode topic:{},deviceId:{} message:{}", message.getTopic(),message.getClientId(), message);
        return context.getDevice(message.getClientId())
                .flatMapMany(deviceOperator -> TopicMatcher.match(message.getTopic())
                                                   .getFunction()
                                                   .apply(message, deviceOperator));
    }

    private Mono<Void> doReply(MessageCodecContext context, TopicPayload reply) {

        if (context instanceof FromDeviceMessageContext) {
            return ((FromDeviceMessageContext) context)
                    .getSession()
                    .send(SimpleMqttMessage
                                  .builder()
                                  .topic(reply.getTopic())
                                  .payload(reply.getPayload())
                                  .build())
                    .then();
        } else if (context instanceof ToDeviceMessageContext) {
            return ((ToDeviceMessageContext) context)
                    .sendToDevice(SimpleMqttMessage
                                          .builder()
                                          .topic(reply.getTopic())
                                          .payload(reply.getPayload())
                                          .build())
                    .then();
        }
        return Mono.empty();

    }

}
